package hzmg.akkacloud.server.core.handler.impl;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.cluster.pubsub.DistributedPubSub;
import akka.cluster.pubsub.DistributedPubSubMediator;
import com.hzmg.akkacloud.common.RemoteConstant;
import com.hzmg.akkacloud.common.request.WorkerHeartbeat;
import lombok.extern.slf4j.Slf4j;

import static hzmg.akkacloud.server.core.handler.WorkerRequestHandler.getWorkerRequestHandler;

/**
 * 处理 Worker 请求
 *
 * @author zbw
 * @since 2022/7/17
 */
@Slf4j
public class WorkerRequestAkkaHandler extends AbstractActor {
    ActorRef mediator = DistributedPubSub.get(getContext().system()).mediator();

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(WorkerHeartbeat.class, this::onReceiveWorkerHeartbeat)
                .matchAny(obj -> log.warn("[WorkerRequestAkkaHandler] receive unknown request: {}.", obj))
                .build();
    }
    /**
     * 处理 Worker 的心跳请求
     * 同时返回最新service信息
     *
     * @param hb 心跳包
     */
    private void onReceiveWorkerHeartbeat(WorkerHeartbeat hb) {
        //new: forward to any registry.
        mediator.tell(new DistributedPubSubMediator.Publish(RemoteConstant.HEARTBEAT_TOPIC, hb), getSelf());
        getSender().tell(getWorkerRequestHandler().onReceiveWorkerHeartbeat(hb), getSelf());
    }

}
